package org.harsh.dispatch;

import cn.hutool.core.codec.Base64;
import cn.hutool.core.date.DateUtil;
import com.ruoyi.common.config.RuoYiConfig;
import com.ruoyi.common.constant.Constants;
import com.ruoyi.common.core.domain.entity.SysDept;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.system.service.ISysConfigService;
import com.ruoyi.system.service.ISysDeptService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.InternetProtocolFamily;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.traffic.ChannelTrafficShapingHandler;
import lombok.extern.slf4j.Slf4j;
import org.harsh.common.Config;
import org.harsh.common.Util;
import org.harsh.transmission.client.NodeClient;
import org.harsh.transmission.common.*;
import org.harsh.transmission.common.codec.*;
import org.harsh.transmission.domain.TransMsg;
import org.harsh.transmission.domain.TransReceiver;
import org.harsh.transmission.domain.TransTask;
import org.harsh.transmission.service.ITransMsgService;
import org.harsh.transmission.service.ITransReceiverService;
import org.harsh.transmission.service.ITransTaskService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.*;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.List;
import java.util.concurrent.*;

@Slf4j
@Component("transmissionTask")
public class TransmissionTask {
    NioEventLoopGroup group = new NioEventLoopGroup();

    @Autowired
    private ITransTaskService transTaskService;
    @Autowired
    private ITransMsgService transMsgService;
    @Autowired
    private ITransReceiverService transReceiverService;
    @Autowired
    private ISysDeptService sysDeptService;
    @Autowired
    private PlainMsgCodec plainMsgCodec;
    @Autowired
    private FileMsgCodec fileMsgCodec;
    @Autowired
    private ISysConfigService sysConfigService;
    @Autowired
    private NodeClient nodeClient;

    private long channelWriteLimit;
    private long channelReadLimit;

    public void run() {
        long channelWriteLimit = 16384l;
        long channelReadLimit = 16384l;
        String tmpWriteStr = sysConfigService.selectConfigByKey("trans.write.channelLimit");
        String tmpReadStr = sysConfigService.selectConfigByKey("trans.read.channelLimit");
        try {
            if(tmpWriteStr != null)
                channelWriteLimit = Long.parseLong(tmpWriteStr);
            if(tmpReadStr != null)
                channelReadLimit = Long.parseLong(tmpReadStr);
        } catch (NumberFormatException e) {
            log.error("【配置】系统参数配置出错");
        }
        CommVars.executorService.submit(()->{
            log.info("Start transmission task.");
            while(true) {
                for(long nodeId : CommVars.nodeTaskQueue.keySet()) {
                    TransTask task = CommVars.takeTask(nodeId);
                    if(task != null) {
                        // 设置为开始传输
                        task.setTaskStatus(1l);
                        task.setStartTime(DateUtil.date());
                        transTaskService.updateTransTask(task);
                        startTaskTrans(task, nodeId);
                        log.info("Start task and trans msg.");
                    }
                }
                TimeUnit.SECONDS.sleep(1);
            }
        });
    }

    private void startTaskTrans(TransTask task, long nodeId) throws InterruptedException, IOException {
        if (task.getTaskType().intValue() == 0) {
            // 单播
            Channel channel = CommVars.getChannel(nodeId, true);
            SysDept sysDept = sysDeptService.selectDeptById(nodeId);
            if(channel == null) {
                log.error("【节点{}】尚未建立连接，正在等待建立连接", nodeId);
                nodeClient.connect(sysDept);
            }
            while(channel == null) {
                TimeUnit.SECONDS.sleep(1);
                channel = CommVars.getChannel(nodeId, true);
            }
            while(!channel.hasAttr(org.harsh.common.Constants.encryptKeyValid)) {
                log.info("Channel does not hava keyValid.");
                TimeUnit.SECONDS.sleep(1);
            }
                // 当前是否已交换密钥
            while(!(boolean)channel.attr(org.harsh.common.Constants.encryptKeyValid).get()) {
                Object obj = channel.attr(org.harsh.common.Constants.encryptKeyValid).get();
                log.info("Channel key is not Valid : {}", obj);
                    // 等待密钥交换
                TimeUnit.SECONDS.sleep(1);
            }

            // todo 自动切换
            channel = CommVars.getChannel(nodeId, NodeStatusCheckTask.canUseTcp(nodeId));
            // todo 添加到对TCP 可用的判断
            Long msgId = task.getMsgId();
            TransMsg msg = transMsgService.selectTransMsgByMsgId(msgId);
            while (!channel.isWritable()) {
                TimeUnit.SECONDS.sleep(50);
            }
            // !important 更新实时跟踪内容
            CommVars.updateTaskStartTime(task, msg.getMsgSize());

            if (msg.getMsgType().intValue() == 0) {
                sendData(channel, msg, task);
//                futureListener(task, future, msgId, nodeId);
            } else {
                String localPath = RuoYiConfig.getProfile();
                // 数据库资源地址
                String filePath = localPath + StringUtils.substringAfter(msg.getMsgData(), Constants.RESOURCE_PREFIX);
                sendFile(channel, msg, task, filePath, nodeId);
            }
        } else if (task.getTaskType().intValue() == 1) {
            // 组播
            System.setProperty("java.net.preferIPv4Stack", "true");
            InetSocketAddress groupAddress = new InetSocketAddress(task.getRecvAddr(), task.getRecvPort());
            SysDept sysDept = sysDeptService.loadLocalDept();
            NetworkInterface ni = null;
            try {
                ni = NetworkInterface.getByInetAddress(InetAddress.getByName(sysDept.getLocalAddr()));
            } catch (SocketException e) {
                e.printStackTrace();
            } catch (UnknownHostException e) {
                e.printStackTrace();
            }
            Enumeration<InetAddress> addresses = ni.getInetAddresses();
            InetAddress localAddress = null;
            while (addresses.hasMoreElements()) {
                InetAddress address = addresses.nextElement();
                if (address instanceof Inet4Address) {
                    localAddress = address;
                }
            }

            long writeLimit = channelWriteLimit;
            long readLimit = channelReadLimit;
            Bootstrap multiBoot = new Bootstrap();
            multiBoot.group(group).channel(NioDatagramChannel.class)
                    .localAddress(localAddress, groupAddress.getPort())
                    // 设置Option 组播
                    .option(ChannelOption.IP_MULTICAST_IF, ni)
                    // 设置Option 重复地址
                    .option(ChannelOption.SO_REUSEADDR, true)
                    // 设置Handler
                    .handler(new ChannelInitializer<NioDatagramChannel>() {
                        @Override
                        protected void initChannel(NioDatagramChannel nioDatagramChannel) throws Exception {
                            ChannelPipeline p = nioDatagramChannel.pipeline();
                            p.addLast(GlobalTraffic.getInstance())
                                    .addLast(new ChannelTrafficShapingHandler(writeLimit, readLimit, 1))
                                    .addLast(new LengthFieldBasedFrameDecoder(14000, 0,4,0,4))
                                    .addLast(new LengthFieldPrepender(4))
                                    .addLast(new DatagramPacketCodec(groupAddress))
                                    .addLast("msgCryptCodec",new PlainCryptMsgPackCodec())
                                    .addLast("msgPackDecoder",new MsgPackDecoder())
                                    .addLast("msgPackEncoder",new MsgPackEncoder())
                                    .addLast("PlainMsgCodec", plainMsgCodec)
                                    .addLast("fileCodec", fileMsgCodec);
//                            p.addLast(new ChannelInboundHandlerAdapter(){
//                                @Override
//                                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
//                                    // todo
//                                    InetSocketAddress address = (InetSocketAddress)ctx.channel().remoteAddress();
//                                    String peerAddr = address.getAddress().getHostAddress();
//                                    log.error("【Channel】连接{}出错-peer Addr :{}", ctx.channel().id(), peerAddr);
//                                    NodeQueue.remove(peerAddr);
//                                    ctx.channel().close();
//                                }
//                            });
                        }
                    });
            ChannelFuture channelFuture = multiBoot.bind(sysDept.getMulticastSendPort()).sync();
            Long msgId = task.getMsgId();
            TransMsg msg = transMsgService.selectTransMsgByMsgId(msgId);

            // !important 更新实时跟踪内容
            CommVars.updateTaskStartTime(task, msg.getMsgSize());
            sendData(task, msg, channelFuture, nodeId);
        } else if (task.getTaskType().intValue() == 2) {
            // 广播
            long writeLimit = channelWriteLimit;
            long readLimit = channelReadLimit;
            SysDept sysDept = sysDeptService.loadLocalDept();
            InetSocketAddress remoteAddress = new InetSocketAddress(task.getRecvAddr(), task.getRecvPort());
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channelFactory(new ChannelFactory<NioDatagramChannel>() {
                        @Override
                        public NioDatagramChannel newChannel() {
                            // 指定协议簇为 IPv4
                            return new NioDatagramChannel(InternetProtocolFamily.IPv4);
                        }
                    })
                    .option(ChannelOption.SO_BROADCAST, true)
                    .handler(new ChannelInitializer<NioDatagramChannel>() {
                        @Override
                        protected void initChannel(NioDatagramChannel ch) throws Exception {
                            // TODO Auto-generated method stub
                            ch.pipeline().addLast(GlobalTraffic.getInstance())
                                    .addLast(new ChannelTrafficShapingHandler(1024, 1024, 1))
                                    .addLast(new LengthFieldBasedFrameDecoder(14000, 0,4,0,4))
                                    .addLast(new LengthFieldPrepender(4))
                                    .addLast(new DatagramPacketCodec(remoteAddress))
                                    .addLast("msgCryptCodec",new PlainCryptMsgPackCodec())
                                    .addLast("msgPackDecoder",new MsgPackDecoder())
                                    .addLast("msgPackEncoder",new MsgPackEncoder())
                                    .addLast("PlainMsgCodec", plainMsgCodec)
                                    .addLast("fileCodec", fileMsgCodec);
                        }
                    });
            ChannelFuture channelFuture = bootstrap.bind(sysDept.getBoardcastSendPort()).sync();
            Long msgId = task.getMsgId();
            TransMsg msg = transMsgService.selectTransMsgByMsgId(msgId);
            // !important 更新实时跟踪内容
            CommVars.updateTaskStartTime(task, msg.getMsgSize());
            sendData(task, msg, channelFuture, nodeId);
        } else if (task.getTaskType().intValue() == 3) {
            // 单向
            SysDept peerNode = sysDeptService.selectDeptById(task.getRecvNode());
            // 单向发送到对方广播端口
            InetSocketAddress remoteAddress = new InetSocketAddress(peerNode.getLocalAddr(), peerNode.getBoardcastPort());
            long writeLimit = channelWriteLimit;
            long readLimit = channelReadLimit;
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channel(NioDatagramChannel.class)
                    .option(ChannelOption.SO_BROADCAST, true)
                    .handler(new ChannelInitializer<NioDatagramChannel>() {
                        @Override
                        protected void initChannel(NioDatagramChannel ch) throws Exception {
                            // TODO Auto-generated method stub
                            ch.pipeline().addLast(GlobalTraffic.getInstance())
                                    .addLast(new ChannelTrafficShapingHandler(writeLimit, readLimit, 1))
                                    .addLast(new LengthFieldBasedFrameDecoder(14000, 0,4,0,4))
                                    .addLast(new LengthFieldPrepender(4))
                                    .addLast(new DatagramPacketCodec(remoteAddress))
                                    .addLast("msgCryptCodec",new PlainCryptMsgPackCodec())
                                    .addLast("msgPackDecoder",new MsgPackDecoder())
                                    .addLast("msgPackEncoder",new MsgPackEncoder())
                                    .addLast("PlainMsgCodec", plainMsgCodec)
                                    .addLast("fileCodec", fileMsgCodec);

                        }
                    });
            ChannelFuture channelFuture = bootstrap.bind(0).sync();
            Long msgId = task.getMsgId();
            TransMsg msg = transMsgService.selectTransMsgByMsgId(msgId);
            // !important 更新实时跟踪内容
            CommVars.updateTaskStartTime(task, msg.getMsgSize());
            sendData(task, msg, channelFuture, nodeId);
        } else {
            log.error("[task-{}]传输类型异常，", task.getTaskId());
        }
    }

    // 发送数据
    private void sendData(Channel channel, TransMsg msg, TransTask task) throws IOException {
        CustomMsg customMsg = new CustomMsg();
        customMsg.setMsgType((byte) org.harsh.common.Constants.MessageType.TRANSMSG.getCode());
        customMsg.setTransType(task.getTaskType().byteValue());
        customMsg.setMsgId(msg.getMsgId());
        customMsg.setFromNode(msg.getFromNode().intValue());
        customMsg.setFromType(msg.getFromType().byteValue());
        customMsg.setFromId(msg.getFromId().intValue());
        if(customMsg.getFromType() == 0 && task.getTaskType().intValue() ==0 ) {
            // 是用户发出的，需要接收用户信息
            TransReceiver receiver = new TransReceiver();
            receiver.setMsgId(msg.getMsgId());
            List<TransReceiver> receiverList = transReceiverService.selectTransReceiverList(receiver);
            customMsg.setRecvId(receiverList.get(0).getRecvId().intValue());
        } else {
            customMsg.setRecvId(0);
        }
        byte[] msgData = msg.getMsgData().getBytes(Config.DEFAULT_CHARSET);
        customMsg.setMsgMd5(Util.byteArrMD5(msgData));
        customMsg.setMsgData(msgData);
        long msgId = msg.getMsgId();
        long nodeId = msg.getFromNode();
        channel.writeAndFlush(customMsg).addListener((ChannelFutureListener) future1 -> {
            if (future1.isSuccess()) {
                // 传输成功
                transTaskService.updateTransTask(task);
                log.info("【节点{}】消息-{}发送成功.", nodeId, msgId);

            } else {
                // 中断传输，传输出现异常
                task.setTaskStatus(2l);
                transTaskService.updateTransTask(task);
                CommVars.loadedTask.remove(task.getTaskId());
                log.info("【节点{}】消息-{}发送异常.", nodeId, msgId);
            }
        });
    }

    // 发送文件
    private void sendFile(Channel channel, TransMsg msg, TransTask task, String filePath, long nodeId) throws InterruptedException, IOException {
        File file = new File(filePath);
        if (!file.exists()) {
            log.error("【传输】MsgId: {} - Filepath [{}] is null.", msg.getMsgId(), filePath);
        } else {
            int recvUserId = 0;
            if(msg.getFromType() == 0 && task.getTaskType().intValue() == 0) {
                // 是用户发出的，需要接收用户信息
                TransReceiver receiver = new TransReceiver();
                receiver.setMsgId(msg.getMsgId());
                List<TransReceiver> receiverList = transReceiverService.selectTransReceiverList(receiver);
                recvUserId = receiverList.get(0).getRecvId().intValue();
            }
            String fileName = file.getName();
            CustomFile customFile = new CustomFile(msg, task, recvUserId, filePath, fileName);
            // todo change file head info
            long startTime = System.currentTimeMillis();
            log.info("Start send file {}:{} to peer, file length : {}, time at {}",
                    customFile.getMsgId(), customFile.getFilePath(),
                    customFile.getFileSize(), startTime);
            channel.writeAndFlush(customFile);

            RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r");
            int byteRead;
            long startPos = customFile.getStartpos();
            byte[] tmpData = new byte[Config.fileChunkSize];
            long length = (int) file.length() / Config.fileChunkSize;
            for (int i = 0; i < length+1; i++) {
                randomAccessFile.seek(startPos);
                if((byteRead = randomAccessFile.read(tmpData)) != -1) {
                    CustomFile customFileData = new CustomFile(
                            org.harsh.common.Constants.MessageType.TRANSFILEDATA.getCode(),
                            customFile.getMsgId(),
                            startPos,
                            Arrays.copyOf(tmpData, byteRead));
                    while(!channel.isWritable()) {
                        TimeUnit.MILLISECONDS.sleep(100);
                    }
                    if(channel.isWritable()) {
                        long finalStartPos = startPos;
//                        log.info("file Trans at :{}, and md5 :{}", startPos, Base64.encode(customFileData.getChunkMd5()));
//                        log.info("file Trans at :{}, and data :{}", startPos, Base64.encode(customFileData.getChunkData()));
                        channel.writeAndFlush(customFileData).addListener((ChannelFutureListener) future-> {
                            if (future.isSuccess()) {
                                CommVars.updateTaskSendOffset(task ,finalStartPos);
                                //task.setTaskOffset();
                                // 传输成功
                                log.info("【节点{}】文件-{}已发送 {}/{} B.", nodeId, msg.getMsgId(), finalStartPos, file.length());
                            } else {
                                // 中断传输，传输出现异常
                                log.info("【节点{}】文件-{}发送异常.", nodeId, msg.getMsgId());
                            }
                        });
                    }
                }
                startPos += byteRead;
            }
            randomAccessFile.close();
            long msgId = msg.getMsgId();
            CustomFile customFileOver = new CustomFile(org.harsh.common.Constants.MessageType.TRANSFILEOVER.getCode(), customFile.getMsgId());
            channel.writeAndFlush(customFileOver).addListener((ChannelFutureListener) future1 -> {
                if (future1.isSuccess()) {
                    // 传输成功
                    CommVars.updateTaskSendOffset(task ,file.length());
                    transTaskService.updateTransTask(task);
                    log.info("【节点{}】消息-{}发送成功.", nodeId, msgId);
                } else {
                    // 中断传输，传输出现异常
                    task.setTaskStatus(2l);
                    transTaskService.updateTransTask(task);
                    CommVars.loadedTask.remove(task.getTaskId());
                    log.info("【节点{}】消息-{}发送异常.", nodeId, msgId);
                }
            });
            //futureListener(task, future, msg.getMsgId(), nodeId);
            long overTime = System.currentTimeMillis();
            long costTime = (overTime - startTime);
            double speed = customFile.getFileSize() / (costTime * 1024 / 1000 );
            log.info("Send file data over to peer, over time: {}, cost Time {} ms, speed {} kB/s", overTime, costTime, speed);
        }
    }


    // 组播、广播、单向发送
    private void sendData(TransTask task, TransMsg msg, ChannelFuture channelFuture, long nodeId) throws IOException, InterruptedException {
        String taskTypeStr;
        switch (task.getTaskType().intValue()) {
            case 1:
                taskTypeStr = "组播";
                break;
            case 2:
                taskTypeStr = "广播";
                break;
            case 3:
                taskTypeStr = "单向";
                break;
            default:
                taskTypeStr = "";
        }
        String finalTaskTypeStr = taskTypeStr;
        if (msg.getMsgType().intValue() == 0) {
            sendData(channelFuture.channel(), msg, task);
        } else {
            String localPath = RuoYiConfig.getProfile();
            // 数据库资源地址
            String filePath = localPath + StringUtils.substringAfter(msg.getMsgData(), Constants.RESOURCE_PREFIX);
            sendFile(channelFuture.channel(), msg, task, filePath, nodeId);
        }
        channelFuture.addListener((ChannelFutureListener) future1 -> {
            if(future1.isDone()) {
                if (future1.isSuccess()) {
                    // 传输成功
                    log.info("【{}】消息{}发送成功.", finalTaskTypeStr, task.getMsgId());
                    task.setTaskStatus(3l);
                    transTaskService.updateTransTask(task);
                    // 从已装载队列里移除 任务 信息
                    CommVars.loadedTask.remove(task.getTaskId());
                    channelFuture.channel().close().await();
                } else {
                    // 中断传输，传输出现异常
                    task.setTaskStatus(2l);
                    transTaskService.updateTransTask(task);
                    // 从已装载队列里移除 任务 信息
                    CommVars.loadedTask.remove(task.getTaskId());
                    channelFuture.channel().close();
                    log.info("【{}】消息{}发送异常.", finalTaskTypeStr, task.getMsgId());
                }
            }
        });
    }
}
